
#include "transfer.hpp"

#include "common.hpp"
#include "items.hpp"


Ping pinger;

// Connector

int Connector::timer(struct timeval *elapsed) const
{
	static struct timeval start;
	struct timeval end;

	if(!elapsed) {
		if(gettimeofday(&start, NULL) < 0)
			return -1;
	} else {
		if(!start.tv_sec || gettimeofday(&end, NULL) < 0)
			return -1;
		elapsed->tv_sec = end.tv_sec - start.tv_sec;		
		elapsed->tv_usec = end.tv_usec - start.tv_usec;
	}
		
	return 0;
}



Connector::PumpStatus Connector::pump_chunk()
{

	IMutexLock<int, PTHREAD_MUTEX_RECURSIVE>& fd_in = source->getFdLock(), fd_out = dest->getFdLock();
	int to_pump = chunk_size_lock.read();

	char buf[to_pump];
	char *end_p, *read_p;
	const char *write_p;

	register int read_b, written_b;
		
	write_p = read_p = buf;
	end_p = buf + to_pump;
		
	// FIXME: Use message non-discard mode?

	while(read_p < end_p) {

		fd_in.aquire();
		
		if((read_b = read(*fd_in, read_p, end_p - read_p)) < 0) {
			error(ErrorReporter::RECOVERABLE, "Reading from source failed");
#ifdef DEBUG
			perror("fd_in_read");
#endif
			fd_in.release();
			return ERROR;
		}

		fd_in.release();

		if(read_b == 0) {
			ErrorReporter::status("Reached End Of File");
			return EMPTY;
		}

		read_p += read_b;

		fd_out.aquire();
		
		// Let writes catch up
		for( ; write_p < read_p; write_p += written_b) {

			if((written_b = write(*fd_out, write_p, read_p - write_p)) < 0) {
				error(ErrorReporter::RECOVERABLE, "Writing to sink failed");
#ifdef DEBUG
			perror("fd_out_read");
#endif
				fd_out.release();
				return ERROR;
			}
		}

		fd_out.release();

		total_transferred_lock.set(total_transferred_lock.read() + to_pump);

		int max_delay = random_bulkiness_delay_lock.read();

		if(max_delay > 0)
			usleep((PrimitiveRandomizer::random_long(true) % max_delay));
	}

	// Still stuff to pump
	return PUMPABLE;
}

bool Connector::isPumping()
{
	return isRunning() && source->isReady() && dest->isReady();
}

void Connector::action(void *)
{

#ifdef DEBUG
	std::cerr << "Waiting for source and destination to report ready\n";
#endif

	while(!source->isReady() || !dest->isReady()) {

		if(!isRunning()) return;
		
		if(have_errors())
			return;
			
		usleep(500000);
	}

	struct timeval t;


#ifdef DEBUG
	std::cerr << "Beginning to pump bytes ...\n";
#endif
	
	while(isRunning()) {
		
		if(have_errors())
			return;
			
		timer(NULL);

		PumpStatus ps = pump_chunk();

		if(ps == EMPTY || ps ==  ERROR) {

#ifdef DEBUG
			std::cerr << "Byte pump stopped\n";
#endif
			if(ps == ERROR) {
				ErrorReporter::error(ErrorReporter::RECOVERABLE, "Data transfer stopped");
#ifdef DEBUG
				perror("byte_pump_error");
#endif				
			}

			// TODO
			// stop();
			return;
		}
		
		timer(&t);

		// Calculate current data rate
		
		double bits = chunk_size_lock.read() * 8;
		double seconds = t.tv_sec + static_cast<double>(t.tv_usec) / 1000000;

		s11n_lock.set(t.tv_sec * 1000000 + t.tv_usec);

		double rate = (seconds > 0) ? (bits / seconds) : bits;
		
		data_rate_lock.set(rate);
	}
}

// TCP Transferer

int TCPtransferer::prepare(volatile int &sockfd, uint16_t &port, struct sockaddr_in *sa)
{
	if((sockfd = socket(AF_INET, SOCK_STREAM, 0)) < 0)
		return -1;
	
	sa->sin_family = AF_INET;
	sa->sin_port = htons(port);
	
	memset(sa->sin_zero, 0, sizeof sa->sin_zero);

	return 0;
}

// TCP destination

void TCPdestination::getReady()
{
	if(isReady()) return;
	ErrorReporter::recover();



	if(prepare(getFdRef(), port, &remote) != 0) {
		error(ErrorReporter::RECOVERABLE, "Unable to setup client socket");
		return;
	}

	remote.sin_addr.s_addr = inet_addr(remote_ip.c_str());

	ErrorReporter::status("Attempting to connect to " + remote_ip);

	if(connect(getFdRef(), (struct sockaddr*)&remote, sizeof(struct sockaddr)) != 0) {
		error(ErrorReporter::RECOVERABLE, "Unable to connect to " + remote_ip);
		return;
	}

	pinger.init(remote_ip.c_str());
	pinger.setup();

	if(pinger.hasErrors()) {
		
		std::cerr << pinger.errorMsg() << std::endl;
	}
	
	pinger.start();

	ErrorReporter::status("Connected to " + remote_ip);

	DataTransferer::setReady(true);
}

// TCP source (server)

void TCPsource::getReady()
{
	if(isReady()) return;
	ErrorReporter::recover();
		
	if(prepare(server_fd, port, &local) != 0) {
		error(ErrorReporter::RECOVERABLE, "Unable to setup server socket");
		return;
	}
		

	local.sin_addr.s_addr = INADDR_ANY;

	int param = 1;
	
	setsockopt(server_fd, SOL_SOCKET, SO_REUSEADDR, &param, sizeof(int));
	
	if(bind(server_fd, (struct sockaddr *)&local, sizeof(struct sockaddr)) != 0) {
		error(ErrorReporter::RECOVERABLE, "Cannot bind to port " +
		      to_string<uint16_t>(port));
		return;
	}
		
	if((listen(server_fd, backlog) != 0)) {
		error(ErrorReporter::RECOVERABLE, "Cannot listen on port " +
		      to_string<uint16_t>(port));
		return;
	}

	// Start accepting connections; see action(void *)
	IThread::start();
}

void TCPsource::action(void *)
{
		
#ifdef DEBUG
	std::cerr << "Accepting connections on " << port << std::endl;
#endif

	int data_fd;
	socklen_t slen = sizeof(struct sockaddr_in);

	ErrorReporter::status("Accepting connections on port " + to_string<uint16_t>(port));

	if((data_fd = accept(server_fd, (struct sockaddr*)&remote, &slen)) < 0) {
		error(ErrorReporter::RECOVERABLE, "Accepting connection on port " + (long int)port);
		return;
	}
		
	DataTransferer::setFd(data_fd);
		
	ErrorReporter::status("Accepted connection from " + connectedWithIP());

	pinger.init(connectedWithIP().c_str());
	pinger.setup();

	if(pinger.hasErrors()) {

		/*app_console_lock.aquire();
		*app_console << Console::note << pinger.errorMsg() << Console::endl;
		app_console_lock.release();
		*/

		std::cerr << pinger.errorMsg() << std::endl;
	}
	
	pinger.start();
	
#ifdef DEBUG
	std::cerr << "Pinging commenced ...\n";
#endif
	
	// Connection is now set up; file descriptor is ready for data transfer
	DataTransferer::setReady(true);
}
